package com.fanxl.flink.project.app;

import com.alibaba.fastjson.JSON;
import com.fanxl.flink.project.domain.Access;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Objects;

/**
 * @description:
 * @author: fanxiaole
 * @date: 2022/3/20 23:35
 */
public class ClickhouseAppV1 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> dataStreamSource = env.readTextFile("data/access.json");
        SingleOutputStreamOperator<Access> cleanStream = dataStreamSource.map(new MapFunction<String, Access>() {
            @Override
            public Access map(String value) throws Exception {
                try {
                    return JSON.parseObject(value, Access.class);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return null;
            }
        }).filter(Objects::nonNull)
                .filter(new FilterFunction<Access>() {
                    @Override
                    public boolean filter(Access value) throws Exception {
                        return "startup".equals(value.getEvent());
                    }
                });
        cleanStream.addSink(JdbcSink.sink("insert into ch_test values(?, ?)",
                (pstmt, x) -> {
                    pstmt.setString(1, x.device);
                    pstmt.setString(2, x.deviceType);
                },
                JdbcExecutionOptions.builder().withBatchSize(10).withBatchIntervalMs(4000).build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:clickhouse://10.215.7.21:8123/flink_study")
        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
        .build()));
        env.execute("ClickhouseAppV1");
    }
}
